
#include "stdafx.h"
#include "ConstDef.h"
#include "ReloadCode.h"
#include "ReloadApp.h"
#include "ReloadPacket.h"
#include "ReloadOverlay.h"
#include "ReloadPacket.h"
#include "ReloadTopologyPluginInterface.h"
#include "ReloadChord.h"
#include "ReloadNetworkDataHandler.h"
#include "ReloadConnectionActor.h"
#include "ReloadDebugPacket.h"
#include "ReloadDebugMgr.h"

namespace reload
{
	ReloadDebugMgr::ReloadDebugMgr(boost::weak_ptr<ReloadOverlay> _pReloadOverlay)
		:pReloadOverlay_(_pReloadOverlay)
	{

	}

	ReloadDebugMgr::~ReloadDebugMgr()
	{

	}

	retT ReloadDebugMgr::init()
	{
		R(asyframe::HandlerBase::init());
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		pUdpHandler_.reset(new asyframe::UdpHandler());
		boost::shared_ptr<boost::asio::ip::udp::endpoint> pLocalEndpoint(new boost::asio::ip::udp::endpoint(
			boost::asio::ip::udp::v4(), pReloadOverlay->listeningIpAddressPort_.port_ + 1));
		pUdpHandler_->setLocalEndpoint(pLocalEndpoint);
		R(pUdpHandler_->attachHandlerOwner(shared_from_this()));
		RETSUCCESS;
	}

	retT ReloadDebugMgr::start()
	{
		R(asyframe::HandlerBase::start());
		R(readOneCmd());
		RETSUCCESS;
	}

	retT ReloadDebugMgr::readOneCmd()
	{
		if(boost::shared_ptr<asyframe::WorkerBase> pWorker = getWorker())
		{
			boost::shared_ptr<asyframe::UdpHandlerContext> pUdpHandlerContext(new asyframe::UdpHandlerContext());
			boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(pWorker));
			pUdpHandlerContext->pBuffer_ = pBuffer;
			boost::shared_ptr<boost::asio::ip::udp::endpoint> pRemoteEndpoint(new boost::asio::ip::udp::endpoint());
			pUdpHandlerContext->pBuffer_->clear();
			pUdpHandlerContext->bytesToTransfer_ = sizeof(int);
			R(pUdpHandler_->asyncReadFrom(pUdpHandlerContext->pBuffer_, pRemoteEndpoint, pUdpHandlerContext,
				boost::bind(&ReloadDebugMgr::onReadCmd, boost::dynamic_pointer_cast<ReloadDebugMgr>(shared_from_this()),
				pUdpHandlerContext, _1, _2, _3)));
		}
		else
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		RETSUCCESS;
	}

	retT ReloadDebugMgr::onReadCmd( boost::shared_ptr<asyframe::UdpHandlerContext> _pUdpHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
	{
		R(IUdpAction::onReadFrom(_pUdpHandlerContext, _bytesTransferred, e, _timeoutCount));
		R(readOneCmd());
		if(!e)
		{
			if(_pUdpHandlerContext && _pUdpHandlerContext->pRemoteEndpoint_)
			{
				if(_pUdpHandlerContext->pBuffer_)
				{
					if(_pUdpHandlerContext->pBuffer_->size() == sizeof(int))
					{
						int cmd;
						R(_pUdpHandlerContext->pBuffer_->read(cmd));
						R(processCmd(cmd, _pUdpHandlerContext->pRemoteEndpoint_));
					}
				}
			}
		}
		else
		{
			LOG_INFO(boost::format( "errorNo %1% : %2% in %3% %4% %5%" ) % e.value() % e.message() % __FUNCTION__ % __FILE__ % __LINE__);
		}
		RETSUCCESS;
	}

	retT ReloadDebugMgr::processCmd( int _cmd, boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		switch(_cmd)
		{
		case ReloadDebugCmd_ChordInfo:R(responseChordInfo(_pRemoteEndpoint));break;
		case ReloadDebugCmd_DataStorageInfo:R(responseDataStorageInfo(_pRemoteEndpoint));break;
		default:break;
		}
		RETSUCCESS;
	}

	retT ReloadDebugMgr::responseChordInfo( boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay || !pReloadOverlay->pConnectionTable_)
		{
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		}
		boost::shared_ptr<ChordRoutingTable> pChordRoutingTable = boost::dynamic_pointer_cast<ChordRoutingTable>(pReloadOverlay->pRoutingTable_);
		if(!pChordRoutingTable)
		{
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		}
		ChordInfoDebugPacket chordInfoDebugPacket;
		chordInfoDebugPacket.selfNodeInfo_.nodeId_ = pReloadOverlay->nodeId_;
		chordInfoDebugPacket.selfNodeInfo_.ipAddressPort_ = pReloadOverlay->listeningIpAddressPort_;
		for(std::size_t i = 0; i < pChordRoutingTable->pNeighborTable_->predecessors_.size() && pChordRoutingTable->pNeighborTable_->predecessors_.size() > 0; ++i)
		{
			NodeInfo nodeInfo;
			nodeInfo.nodeId_ = pChordRoutingTable->pNeighborTable_->predecessors_[i] + pReloadOverlay->nodeId_;
			boost::shared_ptr<ReloadConnectionActor> pReloadConnectionActor = pReloadOverlay->pConnectionTable_->nodeIdIndexes_[nodeInfo.nodeId_].lock();
			if(!pReloadConnectionActor)
				RET(INFO_PROTOCOL_RELOAD_DEBUG_CMD_NOT_SUCCESS);
			nodeInfo.ipAddressPort_ = pReloadConnectionActor->getIpAddressPort();
			chordInfoDebugPacket.predecessors_.array_.push_back(nodeInfo);
		}
		for(std::size_t i = 0; i < pChordRoutingTable->pNeighborTable_->successors_.size() && pChordRoutingTable->pNeighborTable_->successors_.size() > 0; ++i)
		{
			NodeInfo nodeInfo;
			nodeInfo.nodeId_ = pChordRoutingTable->pNeighborTable_->successors_[i] + pReloadOverlay->nodeId_;
			boost::shared_ptr<ReloadConnectionActor> pReloadConnectionActor = pReloadOverlay->pConnectionTable_->nodeIdIndexes_[nodeInfo.nodeId_].lock();
			if(!pReloadConnectionActor)
				RET(INFO_PROTOCOL_RELOAD_DEBUG_CMD_NOT_SUCCESS);
			nodeInfo.ipAddressPort_ = pReloadConnectionActor->getIpAddressPort();
			chordInfoDebugPacket.successors_.array_.push_back(nodeInfo);
		}
		for(std::size_t i = 0; i < pChordRoutingTable->pFingerTable_->fingers_.size() && pChordRoutingTable->pFingerTable_->fingers_[i].size() > 0; ++i)
		{
			chordInfoDebugPacket.fingers_.array_.resize(i + 1);
			for(std::size_t j = 0; j < pChordRoutingTable->pFingerTable_->fingers_[i].size(); ++j)
			{
				NodeInfo nodeInfo;
				nodeInfo.nodeId_ = pChordRoutingTable->pFingerTable_->fingers_[i][j] + pReloadOverlay->nodeId_;
				boost::shared_ptr<ReloadConnectionActor> pReloadConnectionActor = pReloadOverlay->pConnectionTable_->nodeIdIndexes_[nodeInfo.nodeId_].lock();
				if(!pReloadConnectionActor)
					RET(INFO_PROTOCOL_RELOAD_DEBUG_CMD_NOT_SUCCESS);
				nodeInfo.ipAddressPort_ = pReloadConnectionActor->getIpAddressPort();
				chordInfoDebugPacket.fingers_.array_[i].array_.push_back(nodeInfo);
			}
		}
		if(boost::shared_ptr<asyframe::WorkerBase> pWorker = getWorker())
		{
			boost::shared_ptr<asyframe::UdpHandlerContext> pUdpHandlerContext(new asyframe::UdpHandlerContext());
			pUdpHandlerContext->bytesToTransfer_ = chordInfoDebugPacket.size();
			boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(pWorker, pUdpHandlerContext->bytesToTransfer_));
			R(chordInfoDebugPacket.writeToBuffer(pBuffer));
			R(pUdpHandler_->asyncWriteTo(pBuffer, _pRemoteEndpoint, pUdpHandlerContext));
		}
		else
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		RETSUCCESS;
	}

	retT ReloadDebugMgr::responseDataStorageInfo( boost::shared_ptr<boost::asio::ip::udp::endpoint> _pRemoteEndpoint )
	{
		boost::shared_ptr<ReloadOverlay> pReloadOverlay = pReloadOverlay_.lock();
		if(!pReloadOverlay || !pReloadOverlay->pDataStorageMgr_)
		{
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		}
		DataStorageInfoDebugPacket dataStorageInfoDebugPacket;
		const DataStorageMgr::DataStorageMap_::nth_index<1>::type& dataStorageMapIndexByResourceId = pReloadOverlay->pDataStorageMgr_->getDataStorageMapIndexByResourceIdOrder();
		for(DataStorageMgr::DataStorageMap_::nth_index<1>::type::iterator it = dataStorageMapIndexByResourceId.begin(); it != dataStorageMapIndexByResourceId.end(); ++it)
		{
			switch(it->replicaNumber_)
			{
			case 0:dataStorageInfoDebugPacket.nonReplica_++;break;
			case 1:dataStorageInfoDebugPacket.replica1_++;break;
			case 2:dataStorageInfoDebugPacket.replica2_++;break;
			default:dataStorageInfoDebugPacket.unknownReplica_++;break;
			}
		}

		if(boost::shared_ptr<asyframe::WorkerBase> pWorker = getWorker())
		{
			boost::shared_ptr<asyframe::UdpHandlerContext> pUdpHandlerContext(new asyframe::UdpHandlerContext());
			pUdpHandlerContext->bytesToTransfer_ = dataStorageInfoDebugPacket.size();
			boost::shared_ptr<asyframe::IoBuffer> pBuffer(new asyframe::IoBuffer(pWorker, pUdpHandlerContext->bytesToTransfer_));
			R(dataStorageInfoDebugPacket.writeToBuffer(pBuffer));
			R(pUdpHandler_->asyncWriteTo(pBuffer, _pRemoteEndpoint, pUdpHandlerContext));
		}
		else
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		RETSUCCESS;
	}
}
